#include "StreamSocketChannel.h"
#include "EventLoop.h"
#include "InnerType.h"
#include "Manager.h"
#include "MessageHeader.h"
#include "PlatLogger.h"
#include "SocketOpt.h"
#include "Util.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {

    StreamSocketChannel::StreamSocketChannel(EventLoop* loop, int sock_fd, types::connection_id_t connId)
        : Channel(loop, sock_fd, connId), HEADER_LEN(static_cast<int32_t>(sizeof(msg::Header)))
    {
        TRACE_INFO << "create StreamSocketChannel bind loop id:" << loop->id() << " fd:" << sock_fd
                   << " loop type:" << types::IO_MULTI_MODE::toString(loop->mode());
    }

    StreamSocketChannel::~StreamSocketChannel() { ::close(fd()); }

    // void StreamSocketChannel::handleClose()
    // {
    //     Channel::handleClose();
    // }

    void StreamSocketChannel::handleWrite()
    {
        if (isWriting()) {
            ssize_t nRet = SocketOpt::write(fd(), outputBuffer_.peek(), outputBuffer_.readableBytes());
            if (nRet > 0) {
                outputBuffer_.retrieve(static_cast<std::size_t>(nRet));
                if (outputBuffer_.readableBytes() == 0) {
                    /// 数据写完 要马上取消写注册，否则epoll会一直通知写
                    disableWriting();
                    if (writeCompleteCallback_) {
                        RunTask([this]() { writeCompleteCallback_(this); });
                    }
                }
            }
            else {
                SYSTEM_ERROR << "TcpConnection::handleWrite, fd:"_s << fd() << " connId:"_s << connId();
            }
        }
        else {
            TRACE_ERROR << "Connection  is down, no more writing, fd:"_s << fd() << " connId:"_s << connId();
        }
    }

    void StreamSocketChannel::sendMsg(const char* msg, size_t len)
    {
        ssize_t nwrote     = 0;
        size_t  remaining  = len;
        bool    faultError = false;
        if (!isWriting() && outputBuffer_.readableBytes() == 0) {
            nwrote = SocketOpt::write(fd(), msg, len);
            if (nwrote >= 0) {
                remaining = len - static_cast<size_t>(nwrote);
                if (remaining == 0 && writeCompleteCallback_) {
                    RunTask([this]() { writeCompleteCallback_(this); });
                }
            }
            else {  // nwrote < 0
                nwrote = 0;
                if (errno != EWOULDBLOCK) {
                    SYSTEM_ERROR << "StreamSocketChannel::write, error fd:"_s << fd() << " connId:" << connId();
                    if (errno == EPIPE || errno == ECONNRESET) {
                        faultError = true;
                    }
                }
            }
        }
        if (!faultError && remaining > 0) {
            outputBuffer_.append(msg + nwrote, remaining);
            if (!isWriting()) {
                enableWriting();
            }
        }
    }

    /// TCP线程，收到消息之后，就是构建消息，然后投递到订阅消息的线程
    void StreamSocketChannel::handleRead()
    {
        int     savedErrno = 0;
        ssize_t nRet       = inputBuffer_.readFd(fd(), &savedErrno);
        if (nRet > 0) {
            while (true) {
                auto tmpReadableLen = static_cast<uint32_t>(inputBuffer_.readableBytes());
                if (tmpReadableLen < HEADER_LEN) {
                    break;
                }
                const char* bufBegin = inputBuffer_.peek();  /// TCP线程只需要构建消息发走，不需要进行具体的decode操作
                auto        header   = msg::MessageHeader::decode(bufBegin, tmpReadableLen);
                uint32_t    msgTotalLen = header.bodyLen + HEADER_LEN;
                if (msgTotalLen > types::MAX_SINGLE_TCP_MESSAGE_LEN) {
                    TRACE_ERROR << "close session, single tcp messgae too big:"_s << msgTotalLen << " connid:"_s
                                << connId();
                    handleClose(shared::getDisconnectedReason(-1));
                    return;
                }
                if (tmpReadableLen < msgTotalLen) {
                    TRACE_DEBUG << "not enough data, connid:"_s << connId() << " readableLen:"_s << tmpReadableLen;
                    break;
                }
                if (header.id >= shared::InnerMsgId::MAX_INNER_MSG_ID) {
                    QUEUE_MGR.pubIPCDataToBusiThread(bufBegin + sizeof(header), header.bodyLen, header.id, loopId(),
                                                     connId());
                }
                else {
                    TRACE_ERROR << "trans error msg id:"_s << header.id << " connid:"_s << connId();
                }
                inputBuffer_.retrieve(static_cast<size_t>(msgTotalLen));
            }
        }
        else if (nRet == 0) {
            SYSTEM_ERROR << "TcpConnection::handleRead, normal disconnected fd:"_s << fd() << " connId:"_s << connId();
            handleClose(shared::getDisconnectedReason(0));
        }
        else {
            if (savedErrno != EINTR && savedErrno != EAGAIN && savedErrno != EWOULDBLOCK) {
                errno = savedErrno;
                SYSTEM_ERROR << "TcpConnection::handleRead loop id:"_s << loopId();
                handleError(shared::getDisconnectedReason(errno));
            }
        }
    }
}  // namespace foundation
}  // namespace xncc